package com.bungarus.busi;

import com.bungarus.Access;
import com.bungarus.codec.protobuf.MessageProtobuf;
import com.bungarus.model.MessageBuilder;
import com.bungarus.model.MessageType;
import com.bungarus.model.Session;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.log4j.Logger;

import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Handle the messages sent by client
 * Created by tang on 2018/11/19.
 */
@ChannelHandler.Sharable
public class ClientMessageRequestHandler extends ChannelInboundHandlerAdapter {
    Logger logger = Logger.getLogger(ClientMessageRequestHandler.class);

    private final Access access;

    public ClientMessageRequestHandler(Access access) {
        this.access = access;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(msg instanceof MessageProtobuf.Message){
            MessageProtobuf.Message message = (MessageProtobuf.Message)msg;
            switch (message.getHeader().getType()) {
                case MessageType.HEARTBEAT_PING:
                    ctx.writeAndFlush(MessageBuilder.buildHeartbeatPongMessage(message.getHeader().getId(),
                            message.getHeader().getId2()));
                    break;
                case MessageType.ACK:
                    //if ack from client, remove it from client discharge queue
                    DischargeQueue queue = Discharger.getQueueInstance(Discharger.DISCHARGE_NAME_CLIENT);
                    queue.removeFromQueue(Long.valueOf(message.getHeader().getId()));
                    break;
                case MessageType.IM:
                case MessageType.IM_GROUP: {//send message to busi layer
                    InnerClient backBusiNode;
                    do {
                        List<InnerClient> busiClients = access.getInnerClientList().get(Access.CLUSTER_TAG_BUSI);
                        long index = Long.parseLong(message.getHeader().getId()) % busiClients.size();
                        backBusiNode = busiClients.get((int) index);
                    } while (backBusiNode.isShuttingDown());
                    backBusiNode.sendMessage(MessageBuilder.buildAccess2BusiMsg(Access.CLUSTER_TAG_BUSI, message), Access.CLUSTER_TAG_BUSI, ctx);
                    break;
                }
                case MessageType.STATUS: {//send message to transfer layer
                    if(!TimeDamper.dampInMemory(message.getHeader().getFrom())) {//the last changed status of the same end client can only be effective during 3 seconds
                        InnerClient transferNode;
                        do {
                            List<InnerClient> transferClients = access.getInnerClientList().get(Access.CLUSTER_TAG_TRANSFER);
                            long index = Long.parseLong(message.getHeader().getId()) % transferClients.size();
                            transferNode = transferClients.get((int) index);
                        } while (transferNode.isShuttingDown());
                        transferNode.sendMessage(MessageBuilder.buildAccess2TransferMsg(Access.CLUSTER_TAG_TRANSFER, message), Access.CLUSTER_TAG_TRANSFER, ctx);
                    }
                    break;
                }
                case MessageType.LOGOUT: {
                    String sessionId = message.getHeader().getSessionId();
                    Session session = Access.getSessionContainer().findSession(sessionId);
                    if(null != session) {
                        ctx.writeAndFlush(MessageBuilder.buildLogoutSucceed(message.getHeader().getId(),
                                message.getHeader().getId2(),
                                message.getHeader().getFrom(), message.getHeader().getTenantID(),
                                sessionId, message.getHeader().getToken()))
                                .addListener(channelFuture -> {
                                    if (channelFuture.isSuccess()) {
                                        // remove session from access
                                        Access.getSessionContainer().remove(session);
                                        new Authentication(ctx, access).logout(String.valueOf(message.getHeader().getTenantID()), message.getHeader().getFrom());
                                        // remove session from busi
                                        InnerClient client = access.getInnerClients().get(Access.CLUSTER_TAG_TRANSFER);
                                        client.sendMessage(MessageBuilder.buildAccess2TransferMsg(Access.CLUSTER_TAG_TRANSFER, message),
                                                Access.CLUSTER_TAG_TRANSFER);
                                    }
                                });
                    }
                    else{
                        logger.warn("Session is not found when user logout for session id -> " + sessionId);
                    }
                    break;
                }
                default:
                    break;
            }
        }
        else {
            super.channelRead(ctx, msg);
        }
    }

    private final static class TimeDamper {
        private final static long duration = 3L * 1000;
        private final static ConcurrentHashMap<String, Long> damper = new ConcurrentHashMap<>();

        static boolean dampInMemory(String client) {
            Long before = damper.putIfAbsent(client, System.currentTimeMillis());
            if(null == before) {
                return true;
            }
            else {
                Long latest = damper.computeIfPresent(client, (k, last_timestamp) -> {
                    long now = System.currentTimeMillis();
                    if((last_timestamp + duration) <= now) {
                        return now;
                    }
                    else {
                        return last_timestamp;
                    }
                });
                return before.equals(latest);
            }
        }
    }
}